package tasks.io.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tasks.io.core.utils.ByteBufferUtils;
import tasks.io.core.utils.IOUtils;

import java.nio.ByteBuffer;
import java.nio.channels.CompletionHandler;


public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
    private static final Logger log = LoggerFactory.getLogger(ReadCompletionHandler.class);
    private ChannelContext channelContext = null;
    private ByteBuffer readByteBuffer;

    //	private ByteBuffer byteBuffer = ByteBuffer.allocate(ChannelContext.READ_BUFFER_SIZE);

    /**
     * @param channelContext
     * @author tanyaowu
     */
    public ReadCompletionHandler(ChannelContext channelContext) {
        this.channelContext = channelContext;
        this.readByteBuffer = ByteBuffer.allocate(channelContext.getReadBufferSize());
        this.readByteBuffer.order(channelContext.tioConfig.getByteOrder());
    }

    @Override
    public void completed(Integer result, ByteBuffer byteBuffer) {
        if (result > 0) {

            TioConfig tioConfig = channelContext.tioConfig;

            if (tioConfig.statOn) {
                tioConfig.groupStat.receivedBytes.addAndGet(result);
                tioConfig.groupStat.receivedTcps.incrementAndGet();

                channelContext.channelStat.receivedBytes.addAndGet(result);
                channelContext.channelStat.receivedTcps.incrementAndGet();
            }

            channelContext.channelStat.latestTimeOfReceivedByte = SystemTimer.currTime;


            if (tioConfig.getTioListener() != null) {
                try {
                    tioConfig.getTioListener().onAfterReceivedBytes(channelContext, result);
                } catch (Exception e) {
                    log.error("", e);
                }
            }

            readByteBuffer.flip();


            channelContext.decodeRunnable.add(ByteBufferUtils.copy(readByteBuffer));
            channelContext.decodeRunnable.execute();

            if (IOUtils.checkBeforeIO(channelContext)) {
                read();
            }

        } else if (result == 0) {
            log.error("{}, 读到的数据长度为0", channelContext);
            Io.close(channelContext, null, "读到的数据长度为0", CloseCode.READ_COUNT_IS_ZERO);
            return;
        } else if (result < 0) {
            if (result == -1) {
                Io.close(channelContext, null, "对方关闭了连接", CloseCode.CLOSED_BY_PEER);
                return;
            } else {
                Io.close(channelContext, null, "读数据时返回" + result, CloseCode.READ_COUNT_IS_NEGATIVE);
                return;
            }
        }
    }

    private void read() {
        if (readByteBuffer.capacity() == channelContext.getReadBufferSize()) {
            readByteBuffer.position(0);
            readByteBuffer.limit(readByteBuffer.capacity());
        } else {
//			log.error("动态调整了readbuffersize, 原:{} / 新:{}", readByteBuffer.capacity(), channelContext.getReadBufferSize());
            readByteBuffer = ByteBuffer.allocate(channelContext.getReadBufferSize());
        }

        channelContext.asynchronousSocketChannel.read(readByteBuffer, readByteBuffer, this);
    }


    @Override
    public void failed(Throwable exc, ByteBuffer byteBuffer) {
        Io.close(channelContext, exc, "读数据时发生异常: " + exc.getClass().getName(), CloseCode.READ_ERROR);
    }


    public ByteBuffer getReadByteBuffer() {
        return readByteBuffer;
    }
}
